package com.atguigu.app.dws;

import com.atguigu.bean.KeywordBean;
import com.atguigu.func.SplitFunction;
import com.atguigu.uitl.ClickHouseUtil;
import com.atguigu.uitl.MyKafkaUtil;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author hjy
 * @create 2023/3/15 8:47
 */

/**
 * 流量域搜索关键词力度页面浏览各窗口汇总表
 * 数据流:web/app -> 日志服务器(file) -> flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> ClickHouse(DWS)
 * 程 序:Mock -> file -> f1.sh -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK) -> Dws01TrafficKeywordPageViewWindow -> ClickHouse(ZK)
 */
public class Dws01_TrafficKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {
        //todo 1 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        env.enableCheckpointing(5000L);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall-flink/check");
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
//        env.setStateBackend(new HashMapStateBackend());
//        System.setProperty("HADOOP_USER_NAME","atguigu");
        //todo 2 从kafka(dwd_traffic_page_log)读取数据 并给watermark赋值
        String topic="dwd_traffic_page_log";
        String groupID="KeywordPage_demo";
        tableEnv.executeSql("create table page_log(" +
                "`common` MAP<STRING,STRING>," +
                "`page` MAP<STRING,STRING>," +
                "`ts` BIGINT," +
                "rt AS to_timestamp_ltz(ts,3)," +
                " WATERMARK FOR rt AS rt - INTERVAL '5' SECOND" +
                ")"+ MyKafkaUtil.getKafkaDDL(topic,groupID));
        //todo 3 过滤 last_page_id="search" and itemtype=keyword and item is not null
        Table filterTable = tableEnv.sqlQuery("select" +
                "   `page`['item'] item," +
                "   rt" +
                " from page_log " +
                " where `page`['last_page_id']='search'" +
                " and `page`['item_type']='keyword'" +
                " and `page`['item'] is not null");
        tableEnv.createTemporaryView("filter_table",filterTable);
//       有数据 filterTable.execute().print();
        //todo 4 自定义函数分词
        //先注册函数
        tableEnv.createTemporarySystemFunction("SplitFunction", SplitFunction.class);

        //todo 5 使用自定义函数对关键词进行分词处理
        Table keyWordTable = tableEnv.sqlQuery("select " +
                "   rt," +
                "   word" +
                " from filter_table, LATERAL TABLE(SplitFunction(item))");
        tableEnv.createTemporaryView("keyword_table",keyWordTable);
//        有数据 keyWordTable.execute().print();
        //todo 6 分组开窗聚合
        Table resultTable = tableEnv.sqlQuery("SELECT " +
                "   date_format(window_start,'yyyy-MM-dd HH:mm:ss') stt, " +
                "   date_format(window_end,'yyyy-MM-dd HH:mm:ss') edt, " +
                "   word keyword," +
                "   count(*) keyword_count," +
                "   UNIX_TIMESTAMP() ts\n" +
                " FROM TABLE(\n" +
                "    TUMBLE(TABLE keyword_table, DESCRIPTOR(rt), INTERVAL '10' SECONDS))\n" +
                "  GROUP BY word,window_start, window_end");
        tableEnv.createTemporaryView("result_table",resultTable);
//        resultTable.execute().print();
        //todo 写入到clickHouse
        //表转流
        DataStream<KeywordBean> keywordBeanDS = tableEnv.toAppendStream(resultTable, KeywordBean.class);
//        keywordBeanDS.print();
        //自定义一个工具类
        keywordBeanDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_traffic_keyword_page_view_window values(?,?,?,?,?)"));
        //todo 启动程序
        env.execute("Dws01_TrafficKeywordPageViewWindow_demo");

    }
}
